package org.ddpush.im.v1.node.udpconnector;

import java.net.SocketAddress;
import java.net.SocketTimeoutException;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicLong;

import org.ddpush.im.v1.node.ClientMessage;

/**
 * UDP 服务器 接受客户端消息的处理类 http://blog.csdn.net/brok1n/article/details/44694335
 */
public class Receiver implements Runnable {

	/**
	 * UDP 服务器Cannel
	 */
	protected DatagramChannel channel;

	/**
	 * 接受客户端消息的最大消息长度 1024
	 */
	protected int bufferSize = 1024;

	/**
	 * 服务器是否被stop
	 */
	protected boolean stoped = false;
	/**
	 * byte 缓冲区
	 */
	protected ByteBuffer buffer;
	/**
	 * socket 地址
	 */
	private SocketAddress address;

	/**
	 * 添加到消息队列中的消息计数器
	 */
	protected AtomicLong queueIn = new AtomicLong(0);
	/**
	 * 从消息队列中移除的消息计数器
	 */
	protected AtomicLong queueOut = new AtomicLong(0);
	/**
	 * 消息队列 这里的消息队列主要存放 ClientMessage 客户端消息包对象
	 */
	protected ConcurrentLinkedQueue<ClientMessage> mq = new ConcurrentLinkedQueue<ClientMessage>();

	/**
	 * @param channel
	 */
	public Receiver(DatagramChannel channel) {
		this.channel = channel;
	}

	/**
	 * 初始化 给buffer缓冲区分配内存空间
	 */
	public void init() {
		buffer = ByteBuffer.allocate(this.bufferSize);
	}

	/**
	 * 停止服务器
	 */
	public void stop() {
		this.stoped = true;
	}

	/**
	 * Receiver 消息接收线程体
	 */
	public void run() {
		// 如果服务器没有被stop 就一直接收消息
		while (!this.stoped) {
			try {
				// synchronized(enQueSignal){
				// 接收消息
				processMessage();
				// if(mq.isEmpty() == true){
				// enQueSignal.wait();
				// }
				// }
			} catch (Exception e) {
				e.printStackTrace();
			} catch (Throwable t) {
				t.printStackTrace();
			}
		}
	}

	/**
	 * 接收消息
	 * 
	 * @throws Exception
	 */
	protected void processMessage() throws Exception {
		// 清除上次收到的数据
		address = null;
		buffer.clear();
		try {
			// 接收客户端发送的UDP消息
			address = this.channel.receive(buffer);
		} catch (SocketTimeoutException timeout) {

		}
		// 如果消息发送者的地址没找到 服务器就丢弃这个接收到的客户端消息
		if (address == null) {
			try {
				Thread.sleep(1);
			} catch (Exception e) {

			}
			return;
		}
		// limit 设置位当前position position 设置为0
		buffer.flip();
		// 将buffer里接收到的客户端发送过来的数据拷贝到一个新的临时数组
		byte[] swap = new byte[buffer.limit() - buffer.position()];
		System.arraycopy(buffer.array(), buffer.position(), swap, 0, swap.length);
		// 根据这个数据发送者的地址 和发送过来的数据 创建一个ClientMessage对象就是客户端消息包
		ClientMessage m = new ClientMessage(address, swap);
		// 将这个收到的客户端小细胞 放入 消息队列中
		enqueue(m);
		// System.out.println(DateTimeUtil.getCurDateTime()+"
		// r:"+StringUtil.convert(m.getData())+"
		// from:"+m.getSocketAddress().toString());

	}

	/**
	 * 入队 将收到的客户端消息包 放入 消息队列中
	 * 
	 * @param message
	 * @return
	 */
	protected boolean enqueue(ClientMessage message) {
		boolean result = mq.add(message);
		if (result == true) {
			queueIn.addAndGet(1);
		}
		return result;
	}

	/**
	 * 出队 从消息队列中取出一个 客户端消息对象
	 * 
	 * @return
	 */
	protected ClientMessage dequeue() {
		ClientMessage m = mq.poll();
		if (m != null) {
			queueOut.addAndGet(1);
		}
		return m;
	}

	/**
	 * 
	 * 
	 * 取出一个 通过数据类型检查的 客户端消息</br>
	 * 就是从消息队列中取出一个MessageClient 客户端发过来的消息</br>
	 * 并且做一下 消息包的格式验证。
	 * 
	 * @return
	 */

	public ClientMessage receive() {

		ClientMessage m = null;
		while (true) {
			m = dequeue();
			if (m == null) {
				return null;
			}
			if (m.checkFormat() == true) {// 检查包格式是否合法，为了网络快速响应，在这里检查，不在接收线程检查
				return m;
			}
		}
	}
}
